package com.shujia.putdata;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.FileReader;
import java.io.FileWriter;
import java.util.ArrayList;

/*
    使用java模拟hadoop上传文件切分block块的场景。
    现在假设学生数据，每一条是1M，总共是1000M，每128M产生一个block块

    因为后面产生map任务的逻辑我们不好模拟，将128*1.1的思想体现在了上传这一块。后面真正说mapreduce的时候，128*1.1是在产生split切片的时候。
 */
public class PutDataDemo {
    public static void main(String[] args) throws Exception {
        //创建字符缓冲输入流
        BufferedReader br = new BufferedReader(new FileReader("hadoop/data/students.csv"));

        //创建一个索引，记录当前是第几个block块
        int index = 0;

        //创建第一个输出流
        BufferedWriter bw = new BufferedWriter(new FileWriter("hadoop/data/blocks/block-" + index));

        //定义一个变量统计读取的条数
        int rowNum = 0;

        //定义一个集合保存读取到数据
        ArrayList<String> list = new ArrayList<>();

        String line = null;
        while ((line= br.readLine())!=null){
            rowNum++; //行数++

            //将数据添加到集合
            list.add(line);

            //这里用于模拟128*1.1的阈值，当达到这个阈值的时候，开始触发写入block块，写128M
            if(rowNum==140){
                //将每次读取到的140M中的前128M数据写入到一个block中
                //因为我们使用的集合是List集合，而List集合有索引的概念，可以通过get方式通过索引获取数据
                for(int i=index*128;i<index*128+128;i++){
                    if(i!=index*128+127){
                        String info = list.get(i);
                        bw.write(info);
                        bw.newLine();
                    }else {
                        String info = list.get(i);
                        bw.write(info);
                    }
                    bw.flush();
                }
                index++; //下一个block块的编号
                rowNum=12; // 剩余的12M计入到下一次的写入
                bw = new BufferedWriter(new FileWriter("hadoop/data/blocks/block-" + index));
            }
        }

        //将集合中剩余的数据写入到一个block块中
        for(int i=list.size()-rowNum;i<list.size();i++){
            if(i!=list.size()-1){
                String info = list.get(i);
                bw.write(info);
                bw.newLine();
            }else {
                String info = list.get(i);
                bw.write(info);
            }
            bw.flush();
        }

        //释放资源
        bw.close();
        br.close();

        System.out.println("文件上传成功，所有bock块产生完毕！！");


    }
}
